package com.navinfo.opentsp.platform.computing.analysis.service;

import com.navinfo.opentsp.platform.computing.analysis.entity.disconnectedVehicle.PointForDisconnectAnalysis;
import org.apache.spark.Partitioner;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import scala.Tuple2;

import java.io.Serializable;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * 异常断连/长停断连，断连原因判断<br/>
 * 分组排序service
 */
public class DisconnectedVehicleSortGroupService implements Serializable {

    private final Integer parallel;

    public DisconnectedVehicleSortGroupService(Integer parallel) {
        this.parallel = parallel;
    }

    public JavaPairRDD<Long, List<PointForDisconnectAnalysis>> sortAndGroup(JavaRDD<PointForDisconnectAnalysis> pointsRdd) {
        int partitionNum = Optional.ofNullable(this.parallel).orElse(1024);

        return pointsRdd
                // Tuple2(Long -> PointForDisconnectAnalysis)
                .mapPartitionsToPair((PairFlatMapFunction<Iterator<PointForDisconnectAnalysis>, Long, PointForDisconnectAnalysis>) pointIterator ->
                        new Iterator<Tuple2<Long, PointForDisconnectAnalysis>>() {
                            @Override
                            public boolean hasNext() {
                                return pointIterator.hasNext();
                            }

                            @Override
                            public Tuple2<Long, PointForDisconnectAnalysis> next() {
                                PointForDisconnectAnalysis p = pointIterator.next();
                                return new Tuple2<>(p.getTid(), p);
                            }
                        })
                .partitionBy(new TidPartitioner(partitionNum))
                // Tuple2(Long -> List<PointForDisconnectAnalysis>) 分组
                .combineByKey(
                        (Function<PointForDisconnectAnalysis, List<PointForDisconnectAnalysis>>) point -> Stream.of(point).collect(Collectors.toList()),
                        (Function2<List<PointForDisconnectAnalysis>, PointForDisconnectAnalysis, List<PointForDisconnectAnalysis>>) (points, p) -> {
                            points.add(p);
                            return points;
                        },
                        (Function2<List<PointForDisconnectAnalysis>, List<PointForDisconnectAnalysis>, List<PointForDisconnectAnalysis>>) (p1, p2) -> {
                            p1.addAll(p2);
                            return p1;
                        })
                // Tuple2(Long -> PointForDisconnectAnalysis) 排序
                .mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, List<PointForDisconnectAnalysis>>>, Long, List<PointForDisconnectAnalysis>>) tuple2Iterator -> {
                    Iterable<Tuple2<Long, List<PointForDisconnectAnalysis>>> iterator = () -> tuple2Iterator;
                    return StreamSupport.stream(iterator.spliterator(), false)
                            .map(t -> {
                                List<PointForDisconnectAnalysis> collect = t._2.stream()
                                        .filter(dataInfo -> dataInfo.getGpsTime() > 0)
                                        .sorted(Comparator.comparing(PointForDisconnectAnalysis::getGpsTime))
                                        .collect(Collectors.toList());
                                return new Tuple2<>(t._1, collect);
                            }).iterator();
                });
    }

    private static class TidPartitioner extends Partitioner {
        private int partitionNum = 1 << 10;

        public TidPartitioner() {
        }

        public TidPartitioner(int partitionNum) {
            this.partitionNum = partitionNum;
        }

        @Override
        public int numPartitions() {
            return this.partitionNum;
        }

        @Override
        public int getPartition(Object key) {
            try {
                return (int) (Long.parseLong(key.toString()) % partitionNum);
            } catch (NumberFormatException e) {
                e.printStackTrace();
            }
            return 0;
        }
    }
}
